package Data_Analysis_llp

import org.apache.spark.sql.SparkSession

import java.util.Properties

object DataAnalysis3 {
  def main(args: Array[String]): Unit = {
    /*
        DataAnalysis3:分析每个城市不同经营方式的企业发布的招聘信息的数量
        例如:深圳   民营    18
     */
    //  准备sparksql的环境
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("DataAnalysis3")
      .enableHiveSupport()
      .getOrCreate()

    //  准备连接mysql的配置
  val mysql_connect=new Properties()
    mysql_connect.setProperty("user","root")
    mysql_connect.setProperty("password","123456")
    mysql_connect.setProperty("driver","com.mysql.jdbc.Driver")

    //  读取到已经经过处理的数据
    spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/llp?useSSL=false","cleanse",mysql_connect)
      .createOrReplaceTempView("cleanse")

    //  开始数据分析
    //  split(trait,",")[0]   根据逗号分割成集合，使用[0]拿到第一个元素
    /*
        最里面拿到地区，公司特征(trait),还有经营方式(ModeOfOperation)
        然后外面一层根据地区和经营方式分组，并且统计值，得到该经营方式的企业发布的招聘信息数量
     */
    val result=spark.sql(
      """
        |select
        |position,
        |ModeOfOperation,
        |count(*) as number
        |from(
        |select
        |position,
        |trait,
        |split(trait,",")[0] as ModeOfOperation
        |from cleanse
        |) as t1
        |group by position,ModeOfOperation
        |""".stripMargin)

    //  将结果写入mysql
    result.write.mode("overwrite")
      .jdbc("jdbc:mysql://192.168.40.110:3306/llp?useSSL=false","DataAnalysis3",mysql_connect)

    //  将数据写入到hdfs,repartition(1):指定数据文件保存到一个分区
    result.repartition(1).write.mode("overwrite")
      .csv("/llp/DataAnalysis_all/DataAnalysis3.csv")


    //  关闭sparksql的环境
    spark.close()

  }

}
